package com.ccp.learning.udp.core;

import com.ccp.learning.udp.db.DatabaseManager;
import com.ccp.learning.udp.db.RecoveryManager;
import com.ccp.learning.udp.entity.LogEntry;
import com.sun.media.sound.InvalidDataException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

@Slf4j
public class DataProcessor implements Runnable {
    private final BlockingQueue<String> queue;
    private List<LogEntry> currentBatch = new ArrayList<>();
    private volatile boolean running = true;
    @Value("${batch.insert_size:1000}")
    private int batchSize;

    public DataProcessor(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try (Connection conn = DatabaseManager.getConnection()) {
            PreparedStatement stmt = conn.prepareStatement(SQL_INSERT);

            while (running || !queue.isEmpty()) {
                String message = queue.poll(100, TimeUnit.MILLISECONDS);
                if (message != null) {
                    processMessage(message, stmt);
                }

                if (currentBatch.size() >= batchSize) {
                    executeBatch(stmt);
                }
            }

            // 处理最后一批数据
            if (!currentBatch.isEmpty()) {
                executeBatch(stmt);
            }
        } catch (Exception e) {
            RecoveryManager.saveRecoveryData(currentBatch);
            logger.error("数据处理线程异常终止", e);
        }
    }

    private void processMessage(String message, PreparedStatement stmt) {
        try {
            LogEntry log = parseAndValidate(message);
            currentBatch.add(log);
            MonitoringCenter.recordReceived();
        } catch (InvalidDataException e) {
            handleInvalidData(message, e);
        }
    }

    public void stop() {
        this.running = false;
        // 立即持久化当前批次
        if (!currentBatch.isEmpty()) {
            RecoveryManager.saveRecoveryData(currentBatch);
        }
    }

    private LogEntry parseAndValidate(String rawData) throws InvalidDataException {
        try {
            LogEntry entry = new Gson().fromJson(rawData, LogEntry.class);

            // 基础校验
            if (entry.getClientId() == null || entry.getClientId().isEmpty()) {
                throw new InvalidDataException("缺少客户端ID");
            }

            if (entry.getTimestamp() < 1609459200000L) { // 2021-01-01之后
                throw new InvalidDataException("非法时间戳");
            }

            // IP格式校验
            if (!isValidIP(entry.getData().getSrcIp())) {
                throw new InvalidDataException("源IP格式错误");
            }

            return entry;
        } catch (JsonSyntaxException e) {
            throw new InvalidDataException("JSON解析失败", e);
        }
    }

    private boolean isValidIP(String ip) {
        return ip.matches("^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\\.){3}" + "(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$");
    }
}